package cn.doitedu.rtmk.demo4;

import cn.doitedu.rtmk.common.EventBean;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * 规则id: rule-1：
 * 静态画像条件： age:[20,39] , gender=male
 * 实时动态画像标签条件: w行为次数 >=3
 * 触发条件: u行为
 */
public class Rule_1_Calculator implements RuleCalculator {
    Table userProfileTable;
    JSONObject jsonObject;
    ValueState<Integer> rule1State;
    @Override
    public void init(RuntimeContext runtimeContext) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "doitedu:2181");

        Connection hbaseConnection = ConnectionFactory.createConnection(conf);
        userProfileTable = hbaseConnection.getTable(TableName.valueOf("user_profile"));

        jsonObject = new JSONObject();


        // 申请一个keyedState，来记录每个用户的规则1的w行为次数
        rule1State = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-1-state", Integer.class));

    }

    @Override
    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {

        String eventId = eventBean.getEventId();
        /**
         * 规则1
         */
        // 判断事件是否是w，如果是，则做动态画像统计
        if (eventId.equals("w")) {
            rule1State.update(rule1State.value() == null ? 0 : rule1State.value() + 1);
        }

        // 判断事件是否是u，如果是，则要判断该用户的所有规则条件是否已满足
        if (eventId.equals("u")) {


            // 先判断该用户的 “动态画像条件”是否已满足
            if (rule1State.value() != null && rule1State.value() >= 3) {

                // 再去hbase查询静态画像条件是否满足
                Get getParam = new Get(Bytes.toBytes(eventBean.getUid()));
                getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"));
                getParam.addColumn(Bytes.toBytes("f"), Bytes.toBytes("gender"));

                // 调用客户端查询
                Result result = userProfileTable.get(getParam);
                byte[] ageBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("age"));
                byte[] genderBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("gender"));
                int age = Integer.parseInt(Bytes.toString(ageBytes));
                String genderStr = Bytes.toString(genderBytes);


                // 判断该用户的画像标签值，是否满足规则要求的值
                if (age >= 30 && age <= 40 && "male".equals(genderStr)) {

                    jsonObject.put("uid", eventBean.getUid());
                    jsonObject.put("timestamp", eventBean.getTimestamp());
                    jsonObject.put("rule_id", "rule-001");

                    collector.collect(jsonObject.toJSONString());

                }
            }

        }





    }
}
